package com.alibaba.otter.node.etl.autoconfigure;

import com.alibaba.otter.canal.extend.communication.CanalConfigClient;
import com.alibaba.otter.node.common.config.ConfigClientService;
import com.alibaba.otter.node.common.statistics.StatisticsClientService;
import com.alibaba.otter.node.etl.common.datasource.DataSourceService;
import com.alibaba.otter.node.etl.common.db.dialect.DbDialectFactory;
import com.alibaba.otter.node.etl.extract.extractor.*;
import com.alibaba.otter.node.etl.load.loader.LoadContext;
import com.alibaba.otter.node.etl.load.loader.LoadStatsTracker;
import com.alibaba.otter.node.etl.load.loader.OtterLoaderFactory;
import com.alibaba.otter.node.etl.load.loader.db.DataBatchLoader;
import com.alibaba.otter.node.etl.load.loader.db.DbLoadAction;
import com.alibaba.otter.node.etl.load.loader.db.FileLoadAction;
import com.alibaba.otter.node.etl.load.loader.db.interceptor.log.LogLoadInterceptor;
import com.alibaba.otter.node.etl.load.loader.db.interceptor.operation.CanalMysqlInterceptor;
import com.alibaba.otter.node.etl.load.loader.db.interceptor.operation.CanalOracleInterceptor;
import com.alibaba.otter.node.etl.load.loader.db.interceptor.operation.OperationInterceptorFactory;
import com.alibaba.otter.node.etl.load.loader.db.interceptor.sql.SqlBuilderLoadInterceptor;
import com.alibaba.otter.node.etl.load.loader.interceptor.ChainLoadInterceptor;
import com.alibaba.otter.node.etl.load.loader.interceptor.LoadInterceptor;
import com.alibaba.otter.node.etl.select.selector.MessageParser;
import com.alibaba.otter.node.etl.select.selector.OtterSelectorFactory;
import com.alibaba.otter.node.etl.transform.transformer.FileDataTransformer;
import com.alibaba.otter.node.etl.transform.transformer.OtterTransformerFactory;
import com.alibaba.otter.node.etl.transform.transformer.RowDataTransformer;
import com.alibaba.otter.shared.common.utils.extension.ExtensionFactory;
import com.alibaba.otter.shared.common.utils.thread.ExecutorTemplateGetter;
import com.alibaba.otter.shared.etl.model.ObjectData;
import org.springframework.aop.framework.ProxyFactoryBean;
import org.springframework.aop.target.CommonsPool2TargetSource;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import java.util.List;
import java.util.concurrent.ExecutorService;

@Configuration(proxyBeanMethods = false)
public class NodeEtlAutoConfiguration {


    /**
     * extract config bean
     */
    @Bean
    public OtterExtractorFactory otterExtractorFactory(BeanFactory beanFactory) {
        OtterExtractorFactory otterExtractorFactory = new OtterExtractorFactory();
        List<Object> dbBatchExtractor = List.of("freedomExtractor",
                "groupExtractor",
                "databaseExtractor",
                "processorExtractor",
                "fileExtractor",
                "viewExtractor");
        otterExtractorFactory.setBeanFactory(beanFactory);
        otterExtractorFactory.setDbBatchExtractor(dbBatchExtractor);
        return otterExtractorFactory;
    }

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public DatabaseExtractor databaseExtractorTarget(ConfigClientService configClientService,
                                                     DbDialectFactory dbDialectFactory) {
        DatabaseExtractor databaseExtractor = new DatabaseExtractor();
        databaseExtractor.setPoolSize(5);
        databaseExtractor.setConfigClientService(configClientService);
        databaseExtractor.setDbDialectFactory(dbDialectFactory);
        return databaseExtractor;
    }

    @Bean
    public CommonsPool2TargetSource databaseExtractorTargetSource() {
        CommonsPool2TargetSource commonsPool2TargetSource = new CommonsPool2TargetSource();

        commonsPool2TargetSource.setMinIdle(1);
        commonsPool2TargetSource.setMaxSize(-1);
        commonsPool2TargetSource.setTimeBetweenEvictionRunsMillis(60000);
        commonsPool2TargetSource.setMinEvictableIdleTimeMillis(60000);
        commonsPool2TargetSource.setTargetBeanName("databaseExtractorTarget");

        return commonsPool2TargetSource;
    }

    @Bean
    public ProxyFactoryBean databaseExtractor(@Qualifier("databaseExtractorTargetSource") CommonsPool2TargetSource databaseExtractorTargetSource) {
        ProxyFactoryBean proxyFactoryBean = new ProxyFactoryBean();
        proxyFactoryBean.setOptimize(false);
        proxyFactoryBean.setProxyTargetClass(true);
        proxyFactoryBean.setTargetSource(databaseExtractorTargetSource);
        return proxyFactoryBean;
    }

    @Bean
    public FileExtractor fileExtractor(ConfigClientService configClientService,
                                       DbDialectFactory dbDialectFactory) {
        FileExtractor fileExtractor = new FileExtractor();
        fileExtractor.setConfigClientService(configClientService);
        fileExtractor.setDbDialectFactory(dbDialectFactory);
        return fileExtractor;
    }

    @Bean
    public FreedomExtractor freedomExtractor(ConfigClientService configClientService,
                                             DbDialectFactory dbDialectFactory) {
        FreedomExtractor freedomExtractor = new FreedomExtractor();
        freedomExtractor.setConfigClientService(configClientService);
        freedomExtractor.setDbDialectFactory(dbDialectFactory);
        return freedomExtractor;
    }

    @Bean
    public ViewExtractor viewExtractor(ConfigClientService configClientService,
                                       DbDialectFactory dbDialectFactory) {
        ViewExtractor viewExtractor = new ViewExtractor();
        viewExtractor.setConfigClientService(configClientService);
        viewExtractor.setDbDialectFactory(dbDialectFactory);
        return viewExtractor;
    }

    @Bean
    public GroupExtractor groupExtractor(ConfigClientService configClientService,
                                         DbDialectFactory dbDialectFactory) {
        GroupExtractor groupExtractor = new GroupExtractor();
        groupExtractor.setConfigClientService(configClientService);
        groupExtractor.setDbDialectFactory(dbDialectFactory);
        return groupExtractor;
    }

    @Bean
    public ProcessorExtractor processorExtractor(ConfigClientService configClientService,
                                                 DbDialectFactory dbDialectFactory,
                                                 ExtensionFactory extensionFactory,
                                                 DataSourceService dataSourceService,
                                                 ExecutorTemplateGetter executorTemplateGetter) {
        ProcessorExtractor processorExtractor = new ProcessorExtractor();
        processorExtractor.setExtensionFactory(extensionFactory);
        processorExtractor.setDataSourceService(dataSourceService);
        processorExtractor.setExecutorTemplateGetter(executorTemplateGetter);
        processorExtractor.setConfigClientService(configClientService);
        processorExtractor.setDbDialectFactory(dbDialectFactory);
        return processorExtractor;
    }

    /**
     * select bean config
     */
    @Bean
    public MessageParser messageParser(ConfigClientService configClientService,DbDialectFactory dbDialectFactory) {
        MessageParser messageParser = new MessageParser();
        messageParser.setConfigClientService(configClientService);
        messageParser.setDbDialectFactory(dbDialectFactory);
        return messageParser;
    }

    @Bean
    public OtterSelectorFactory otterSelectorFactory() {
        return new OtterSelectorFactory();
    }


    /**
     * Transform bean config
     */
    @Bean
    public RowDataTransformer rowDataTransformer(DbDialectFactory dbDialectFactory) {
        RowDataTransformer rowDataTransformer = new RowDataTransformer();
        rowDataTransformer.setDbDialectFactory(dbDialectFactory);
        return rowDataTransformer;
    }

    @Bean
    public FileDataTransformer fileDataTransformer() {
        return new FileDataTransformer();
    }

    @Bean
    @ConditionalOnBean(ConfigClientService.class)
    public OtterTransformerFactory otterTransformerFactory(RowDataTransformer rowDataTransformer,
                                                           FileDataTransformer fileDataTransformer,
                                                           ConfigClientService configClientService) {
        OtterTransformerFactory otterTransformerFactory = new OtterTransformerFactory();
        otterTransformerFactory.setConfigClientService(configClientService);
        otterTransformerFactory.setRowDataTransformer(rowDataTransformer);
        otterTransformerFactory.setFileDataTransformer(fileDataTransformer);

        return otterTransformerFactory;
    }

    /**
     * load bean config
     */
    @Bean
    public LoadStatsTracker loadStatsTracker() {
        return new LoadStatsTracker();
    }

    @Bean
    public LogLoadInterceptor dbLogLoadInterceptor() {
        LogLoadInterceptor logLoadInterceptor = new LogLoadInterceptor();
        logLoadInterceptor.setDump(true);
        return logLoadInterceptor;
    }

    @Bean
    public SqlBuilderLoadInterceptor sqlBuilderLoadInterceptor(DbDialectFactory dbDialectFactory) {
        SqlBuilderLoadInterceptor sqlBuilderLoadInterceptor = new SqlBuilderLoadInterceptor();
        sqlBuilderLoadInterceptor.setDbDialectFactory(dbDialectFactory);
        return sqlBuilderLoadInterceptor;
    }


    @Bean
    public CanalMysqlInterceptor canalMysqlInterceptor(ConfigClientService configClientService) {
        CanalMysqlInterceptor canalMysqlInterceptor = new CanalMysqlInterceptor();
        canalMysqlInterceptor.setGlobalIdCount(1000);
        canalMysqlInterceptor.setInnerIdCount(300);
        canalMysqlInterceptor.setConfigClientService(configClientService);
        return canalMysqlInterceptor;
    }

    @Bean
    public CanalOracleInterceptor canalOracleInterceptor(ConfigClientService configClientService) {
        CanalOracleInterceptor canalOracleInterceptor = new CanalOracleInterceptor();
        canalOracleInterceptor.setGlobalIdCount(1000);
        canalOracleInterceptor.setInnerIdCount(300);
        canalOracleInterceptor.setConfigClientService(configClientService);
        return canalOracleInterceptor;
    }


    @Bean
    @SuppressWarnings("unchecked")
    public OperationInterceptorFactory operationInterceptorFactory(DbDialectFactory dbDialectFactory,
            CanalOracleInterceptor canalOracleInterceptor,
                                                                   CanalMysqlInterceptor canalMysqlInterceptor) {
        OperationInterceptorFactory operationInterceptorFactory = new OperationInterceptorFactory();
        operationInterceptorFactory.setMysqlInterceptors(new LoadInterceptor[]{canalMysqlInterceptor});
        operationInterceptorFactory.setOracleInterceptors(new LoadInterceptor[]{canalOracleInterceptor});
        operationInterceptorFactory.setDbDialectFactory(dbDialectFactory);
        return operationInterceptorFactory;
    }

    @SuppressWarnings("unchecked")
    @Bean
    public ChainLoadInterceptor dbLoadInterceptor(SqlBuilderLoadInterceptor sqlBuilderLoadInterceptor,
                                                  LogLoadInterceptor dbLogLoadInterceptor,
                                                  OperationInterceptorFactory operationInterceptorFactory) {
        ChainLoadInterceptor chainLoadInterceptor = new ChainLoadInterceptor();
        List<LoadInterceptor<?, ?>> interceptors = List.of(
                sqlBuilderLoadInterceptor,
                operationInterceptorFactory,
                dbLogLoadInterceptor);
        List<LoadInterceptor<LoadContext, ObjectData>> specificInterceptors = interceptors.stream()
                .map(interceptor -> (LoadInterceptor<LoadContext, ObjectData>) interceptor)
                .toList();
        chainLoadInterceptor.setInterceptors(specificInterceptors);
        return chainLoadInterceptor;
    }

    @Bean
    @SuppressWarnings({"unchecked", "rawtypes"})
    public DataBatchLoader dataBatchLoader(ChainLoadInterceptor dbLoadInterceptor,
                                           @Qualifier("executorService") ExecutorService executorService,
                                           BeanFactory beanFactory,
                                           ConfigClientService configClientService) {
        DataBatchLoader dataBatchLoader = new DataBatchLoader();
        dataBatchLoader.setDbInterceptor((LoadInterceptor) dbLoadInterceptor);
        dataBatchLoader.setExecutorService(executorService);
        dataBatchLoader.setBeanFactory(beanFactory);
        dataBatchLoader.setConfigClientService(configClientService);
        return dataBatchLoader;
    }

    @Bean
    public OtterLoaderFactory otterLoaderFactory(DataBatchLoader dataBatchLoader,
                                                 LoadStatsTracker loadStatsTracker,
                                                 StatisticsClientService statisticsClientService) {
        OtterLoaderFactory otterLoaderFactory = new OtterLoaderFactory();
        otterLoaderFactory.setDataBatchLoader(dataBatchLoader);
        otterLoaderFactory.setLoadStatsTracker(loadStatsTracker);
        otterLoaderFactory.setStatisticsClientService(statisticsClientService);
        return otterLoaderFactory;
    }


    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public DbLoadAction dbLoadActionTarget(ChainLoadInterceptor dbLoadInterceptor,
                                           ConfigClientService configClientService,
                                           LoadStatsTracker loadStatsTracker,
                                           CanalConfigClient canalConfigClient,
                                           DbDialectFactory dbDialectFactory) {
        DbLoadAction dbLoadAction = new DbLoadAction();
        dbLoadAction.setRetry(3);
        dbLoadAction.setRetryWait(3000);
        dbLoadAction.setPoolSize(5);
        dbLoadAction.setConfigClientService(configClientService);
        dbLoadAction.setLoadStatsTracker(loadStatsTracker);
        dbLoadAction.setCanalConfigClient(canalConfigClient);
        dbLoadAction.setInterceptor(dbLoadInterceptor);
        dbLoadAction.setDbDialectFactory(dbDialectFactory);
        return dbLoadAction;
    }


    @Bean
    public CommonsPool2TargetSource dbLoadActionTargetSource() {
        CommonsPool2TargetSource commonsPool2TargetSource = new CommonsPool2TargetSource();

        commonsPool2TargetSource.setMinIdle(1);
        commonsPool2TargetSource.setMaxSize(-1);
        commonsPool2TargetSource.setTimeBetweenEvictionRunsMillis(60000);
        commonsPool2TargetSource.setMinEvictableIdleTimeMillis(60000);
        commonsPool2TargetSource.setTargetBeanName("dbLoadActionTarget");

        return commonsPool2TargetSource;
    }

    @Bean
    public ProxyFactoryBean dbLoadAction(@Qualifier("dbLoadActionTargetSource") CommonsPool2TargetSource dbLoadActionTargetSource) {
        ProxyFactoryBean proxyFactoryBean = new ProxyFactoryBean();
        proxyFactoryBean.setOptimize(false);
        proxyFactoryBean.setProxyTargetClass(true);
        proxyFactoryBean.setTargetSource(dbLoadActionTargetSource);

        return proxyFactoryBean;
    }


    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public FileLoadAction fileLoadActionTarget(ConfigClientService configClientService,
                                               LoadStatsTracker loadStatsTracker) {
        FileLoadAction fileLoadAction = new FileLoadAction();
        fileLoadAction.setRetry(3);
        fileLoadAction.setPoolSize(5);
        fileLoadAction.setConfigClientService(configClientService);
        fileLoadAction.setLoadStatsTracker(loadStatsTracker);
        return fileLoadAction;
    }

    @Bean
    public CommonsPool2TargetSource fileLoadActionTargetSource() {
        CommonsPool2TargetSource commonsPool2TargetSource = new CommonsPool2TargetSource();

        commonsPool2TargetSource.setMinIdle(1);
        commonsPool2TargetSource.setMaxSize(-1);
        commonsPool2TargetSource.setTimeBetweenEvictionRunsMillis(60000);
        commonsPool2TargetSource.setMinEvictableIdleTimeMillis(60000);
        commonsPool2TargetSource.setTargetBeanName("fileLoadActionTarget");

        return commonsPool2TargetSource;
    }


    @Bean
    public ProxyFactoryBean fileLoadAction(@Qualifier("fileLoadActionTargetSource") CommonsPool2TargetSource fileLoadActionTargetSource) {
        ProxyFactoryBean proxyFactoryBean = new ProxyFactoryBean();
        proxyFactoryBean.setOptimize(false);
        proxyFactoryBean.setProxyTargetClass(true);
        proxyFactoryBean.setTargetSource(fileLoadActionTargetSource);

        return proxyFactoryBean;
    }
}
